package com.abcdefg.demo.publishConfirm;

import com.abcdefg.demo.constants.Constants;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

import static com.abcdefg.demo.constants.Constants.PUBLISH_CONFIRMS_QUEUE3;

public class PublisherConfirms {
    static int MessageSize=200;
    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.Host);
        factory.setPort(Constants.Port);
        factory.setVirtualHost(Constants.VirtualHost);
        factory.setUsername(Constants.UserName);
        factory.setPassword(Constants.Password);
        return factory.newConnection();
    }
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //单独确认
//        publishMessagesIndividually();
        //批量确认
        publishMessagesInBatch();
        //异步确认
        handlePublishConfirmsAsynchronously();
    }

    private static void handlePublishConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()) {
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();
            channel.queueDeclare(PUBLISH_CONFIRMS_QUEUE3,true,false,false,null);
            //有序集合,元素按照⾃然顺序进⾏排序,存储未confirm消息序号
             SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {

                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple) {
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple) {
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }
                }
            });
            //发送消息
            long startTime = System.currentTimeMillis();
            for (int i = 0; i < MessageSize; i++) {
                String msg="handlePublishConfirmsAsynchronously(异步确认消息):"+i;
                long nextPublishSeqNo = channel.getNextPublishSeqNo();
                 // System.out.println("消息序号:"+ nextPublishSeqNo);
                 channel.basicPublish("",PUBLISH_CONFIRMS_QUEUE3 , null, msg.getBytes());
                 //将序号存⼊集合中
                 confirmSet.add(nextPublishSeqNo);
            }
            while(!confirmSet.isEmpty()){
                Thread.sleep(10 );
            }
            long endTime = System.currentTimeMillis();
            System.out.printf("发送消息的数量：%d,消耗的时间：%d\n",MessageSize,endTime-startTime);
        }
    }

    private static void publishMessagesInBatch() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();
            //声明队列
            channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE2,true,false,false,null);
            //发送消息
            long startTime = System.currentTimeMillis();
            int batchSize=100;
            int count=0;
            for (int i = 0; i < MessageSize; i++) {
                String msg="publishMessagesInBatch(批量确认消息):"+i;
                channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE2,null,msg.getBytes());
                count++;
                if(count==batchSize){
                    channel.waitForConfirms(2000);
                    count=0;
                }
            }
            if(count>0){
                channel.waitForConfirms(2000);
            }
            long endTime = System.currentTimeMillis();
            System.out.printf("发送消息的数量：%d,消耗的时间：%d\n",MessageSize,endTime-startTime);
        }
    }

    private static void publishMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();
            //声明队列
            channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE1,true,false,false,null);
            //发送消息
            long startTime = System.currentTimeMillis();
            for (int i = 0; i < MessageSize; i++) {
                String msg="publishMessagesIndividually(同步确认消息):"+i;
                channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE1,null,msg.getBytes());
                channel.waitForConfirms(2000);
            }
            long endTime = System.currentTimeMillis();
            System.out.printf("发送消息的数量：%d,消耗的时间：%d\n",MessageSize,endTime-startTime);
        }
    }
}
